package com.common.serialize.impl;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;

import com.common.entity.RPCRequest;
import com.common.entity.RPCResponse;
import com.common.exception.SerializeException;
import com.common.serialize.CommonSerializer;
import com.common.serialize.SerializerCode;
import lombok.extern.slf4j.Slf4j;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;

/**
 * @author Zixiang Hu
 * @description 实现Kryo序列化
 * @create 2021-02-27-18:28
 */
@Slf4j
public class KryoSerializer implements CommonSerializer {

    /**
     * Kryo不是线程安全类，为了避免多线程情况下出现错误，在序列化时，每个线程使用自己的线程特有对象进行序列化
     */
    private static final ThreadLocal<Kryo> kryoThreadLocal = ThreadLocal.withInitial(() -> {
        Kryo kryo = new Kryo();
        kryo.register(RPCRequest.class);
        kryo.register(RPCResponse.class);
        kryo.setReferences(true);
        kryo.setRegistrationRequired(false);
        return kryo;
    });

    @Override
    public byte[] serialize(Object obj) {
        try (ByteArrayOutputStream bos = new ByteArrayOutputStream(); Output output = new Output(bos)) {
            // 获得线程特有对象
            Kryo kryo = kryoThreadLocal.get();
            kryo.writeObject(output, obj);
            // 使用完之后记得清除，防止内存泄漏
            kryoThreadLocal.remove();
            return output.toBytes();
        } catch (Exception e) {
            log.error("KryoSerializer的serialize发送异常");
            throw new SerializeException("KryoSerializer的serialize发送异常");
        }
    }

    @Override
    public Object deserialize(byte[] bytes, Class<?> clazz) {
        try (ByteArrayInputStream bis = new ByteArrayInputStream(bytes); Input input = new Input(bis)) {
            Kryo kryo = kryoThreadLocal.get();
            Object o = kryo.readObject(input, clazz);
            kryoThreadLocal.remove();
            return o;
        } catch (Exception e) {
            log.error("KryoSerializer的deserialize发送异常");
            throw new SerializeException("KryoSerializer的deserialize发送异常");
        }
    }

    @Override
    public int getCode() {
        return SerializerCode.valueOf("KRYO").getCode();
    }
}
